package com.atguigu.sparkcore.rdd.kvs

import com.atguigu.sparkcore.util.MySparkContextUtil
import org.apache.spark.{Partitioner, SparkConf, SparkContext}

/**
 * 自定义分区器
 * author 剧情再美终是戏
 * mail 13286520398@163.com
 * date 2020/1/6 21:18
 * version 1.0
 **/
object MyPartitions {


  def main(args: Array[String]): Unit = {

    // 获取sparkContext
    val sc = MySparkContextUtil.get(args)

    // 创建rdd
    val list = Array((1, "a"), (30, "a"), (10, "a"), (21, "a"), (90, "a"), (7, "a"), (4, "a"))
    val rdd = sc.makeRDD(list, 2)

    // 转换rdd
    val result = rdd.partitionBy(MyPartition(2))

    // 输出
    println(result.collect().mkString(","))

    // 关闭资源
    MySparkContextUtil.close(sc)
  }

  /**
   * 自定义分区器
   *
   * @param num 自定义分区数<br>
   *            1、主要重写numPartitions和getPartition方法<br>
   *            2、hashCode，equals方法也要重写，因为当判断两个rdd之间的分区信息时，需要用到这两个方法，
   *            如果这两个方法不重写，可以会造成，同一个分区器，需要构建多次，影响性能
   */
  case class MyPartition(num: Int) extends Partitioner {
    override def numPartitions: Int = num

    override def getPartition(key: Any): Int = key match {
      case key: Int => (key % num).abs
      case _ => 0
    }

    /**
     * TODO 1、hashCode， equals这两个方法，在自定义分区器时，不是必需的
     * TODO 2、如果两个方法定义了，在shuffle过程中，如果使用了同一个的分区器，就可以减少本次不必要的shuffle操作
     *    rdd.reduceByKey(new MyPartition(2)).reduceByKey(new MyPartition(2))
     * 像上面这种多次shuffle,而使用了同样功能的分区器，其实可以省掉不必要的shuffle的
     * 如果需要省掉这不必要的shuffle操作，就需要执行以下的操作：
     * TODO 2.1、hashCode(): Int = num
     * TODO 2.2、 equals方法，如果分区的个数一样，返回true
     */
    override def hashCode(): Int = num

    override def equals(obj: scala.Any): Boolean = obj match {
      case obj: MyPartition => obj.num == num
      case _ => false
    }
  }

}
